package hydrap2p.proxy.manager;

import hydrap2p.ServiceLocation;
import hydrap2p.logging.Level;
import hydrap2p.manager.ModuleManagerImpl;

import java.io.BufferedInputStream;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.Socket;
import java.util.Map;
import java.util.TreeMap;

public class ManagerConnectionProcessor implements Runnable {
	private final static byte CMD_PEERID=1;
	private final static byte CMD_FOUNDMANAGER=2;
	private final static byte CMD_OPENCALLBACK=3;
	private final static byte CMD_REPLYCALLBACK=4;
	private final static byte CMD_ERRORCALLBACK=5;
	private final static byte CMD_SENDDATA=8;
	private final static byte CMD_RETURNDATA=9;
	private final static byte CMD_CLOSEDATA=10;
	
	
	private final Socket socket;
	
	private ActiveConnections connections;
	private boolean running;
	
	private Integer peerID;
	private StreamMultiplexer streams;
	
	public ManagerConnectionProcessor(Socket socket)
	{
		this.socket=socket;
		this.connections=new ActiveConnections(this);
		peerID=null;
	}
	
	@Override
	public void run()
	{
		InputStream socketinput;
		OutputStream socketoutput;
		running=true;
		try {
			socketinput=new BufferedInputStream(socket.getInputStream(), 10240);
			socketoutput=socket.getOutputStream();
			streams=new StreamMultiplexer(socketoutput);
			
			sendInstanceID(socketoutput);
			
			NetworkReader objectinput=new NetworkReader(socketinput);
			
			int dataread=0;							// how much was read for this last call to read()
			int streamid=-1;						// which stream this current block refers to
			int streamsize=0;						// how much left to read in the current stream block
			byte[][] packetdata=new byte[256][];
			int[] packetread=new int[256];
			int[] commandsize=new int[256];
			byte[] command=new byte[256];
			boolean waswaiting=false;
			
			ServiceLocation templocation;
			ServiceLocation templocation2;
			byte[] tempdata;

			while (dataread!=-1)
			{
				// if we are not currently parsing a stream block
				if (streamid==-1)
				{
					// parse out the stream header
					streamid=objectinput.readByte();
					if (streamid<0) streamid=256+streamid;		// turn it into an unsigned byte
					streamsize=objectinput.readInt();
					if (packetdata[streamid]==null)				// if this is the beginning of a new command packet
					{
						command[streamid]=objectinput.readByte();
						commandsize[streamid]=objectinput.readInt();
						packetdata[streamid]=new byte[commandsize[streamid]];
						packetread[streamid]=0;
						streamsize-=5;
					}
				}
				// while we haven't read the entire stream block yet
				while (streamsize>0)
				{
					dataread=socketinput.read(packetdata[streamid], packetread[streamid], streamsize);
	
					if (dataread>0)
					{
						packetread[streamid]+=dataread;
						streamsize-=dataread;
					}
					if (dataread<0)
						break;
				}
				// we've read the entire stream block, do we have the entire command block yet
				if (packetread[streamid]<commandsize[streamid])
				{
					//ProxyLogging.log(Level.DEBUG, "ProxyManager - ManagerConnectionProcessor -> "+peerID, "Waiting to read packet OP"+command[streamid]+" "+packetread[streamid]+" / "+commandsize[streamid]);
					streamid=-1;		// prepare to read another stream block
					waswaiting=true;
					continue;
				}
				if (waswaiting)
				{
					//ProxyLogging.log(Level.DEBUG, "ProxyManager - ManagerConnectionProcessor -> "+peerID, "Finished waiting to read packet OP"+command[streamid]+" "+packetread[streamid]+" / "+commandsize[streamid]);
				}
				packetread[streamid]=-1;
				waswaiting=false;
				
				NetworkReader packetinput = new NetworkReader(new ByteArrayInputStream(packetdata[streamid]));
				switch(command[streamid])
				{
				case CMD_PEERID:
					this.peerID=packetinput.readInt();
					//ProxyLogging.log(Level.DEBUG, "ProxyManager - ManagerConnectionProcessor -> "+peerID, "Received incoming PeerID: "+peerID);
					ManagerConnectionManager.getInstance().setPeerID(peerID, this);
					break;
					
				case CMD_FOUNDMANAGER:
					templocation=packetinput.readServiceLocation();
					//ProxyLogging.log(Level.DEBUG, "ProxyManager - ManagerConnectionProcessor -> "+peerID, "Received incoming Manager location: "+templocation);
					if (Server.isRunning())
						Server.connectToManager(peerID, templocation);
					break;
					
				case CMD_OPENCALLBACK:
					templocation=packetinput.readServiceLocation();
					//ProxyLogging.log(Level.DEBUG, "ProxyManager - ManagerConnectionProcessor -> "+peerID, "Received incoming callback request, calling "+templocation);
					try {
						templocation2=ServiceTunnels.getInstance().createConnection(peerID, templocation);
						//ProxyLogging.log(Level.DEBUG, "ProxyManager - ManagerConnectionProcessor -> "+peerID, "Successful callback created, from "+templocation2+" to "+templocation);
						
						try {
							replyCallbackPort(templocation, templocation2);
						}
						catch (IOException e) {
							// could not reply with port
							ProxyLogging.log(Level.ERROR, "ProxyManager - ManagerConnectionProcessor -> "+peerID, "Failed to tell callback requestor about his port, from "+templocation2+" to "+templocation);
							ServiceTunnels.getInstance().removeTunnel(templocation.getPort());
						}
					}
					catch (IOException e) {
						ProxyLogging.log(Level.ERROR, "ProxyManager - ManagerConnectionProcessor -> "+peerID, "Failed to create tunnel to satisfy callback request, to "+templocation);
						// failure to establish tunnel
						errorCallbackPort(templocation);
					}
					break;
					
				case CMD_REPLYCALLBACK:
					templocation=packetinput.readServiceLocation();
					templocation2=packetinput.readServiceLocation();
					//ProxyLogging.log(Level.DEBUG, "ProxyManager - ManagerConnectionProcessor -> "+peerID, "Successful callback request, from "+templocation2+" to "+templocation);
					NetworkOutstandingCallbacks.finishFor(templocation, templocation2);
					break;
					
				case CMD_ERRORCALLBACK:
					templocation=packetinput.readServiceLocation();
					ProxyLogging.log(Level.ERROR, "ProxyManager - ManagerConnectionProcessor -> "+peerID, "Error in callback request, to "+templocation);
					NetworkOutstandingCallbacks.errorFor(templocation);
					break;
					
				case CMD_SENDDATA:
					templocation=packetinput.readServiceLocation();
					templocation2=packetinput.readServiceLocation();
					tempdata=packetinput.readBytes();
					//ProxyLogging.log(Level.DEBUG, "ProxyManager - ManagerConnectionProcessor -> "+peerID, "Incoming data, from "+templocation+" to "+templocation2+" of "+tempdata.length);
					try {
						connections.sendData(templocation, templocation2, tempdata);
					}
					catch (Exception activeconnectionexception)
					{
						ProxyLogging.log(Level.ERROR, "ProxyManager - ManagerConnectionProcessor -> "+peerID, "Failed to send data to "+templocation2);
						closeDataConnection(templocation, templocation2);
					}
					break;
					
				case CMD_RETURNDATA:
					templocation=packetinput.readServiceLocation();
					templocation2=packetinput.readServiceLocation();
					tempdata=packetinput.readBytes();
					//ProxyLogging.log(Level.DEBUG, "ProxyManager - ManagerConnectionProcessor -> "+peerID, "Incoming returned data, from "+templocation+" to "+templocation2+" of "+tempdata.length);
					ServiceTunnels.getInstance().returnData(peerID, templocation, templocation2, tempdata);
					break;
					
				case CMD_CLOSEDATA:
					templocation=packetinput.readServiceLocation();
					templocation2=packetinput.readServiceLocation();
					//ProxyLogging.log(Level.DEBUG, "ProxyManager - ManagerConnectionProcessor -> "+peerID, "Closed data connection, from "+templocation+" to "+templocation2);
					ServiceTunnels.getInstance().closeDataConnection(peerID, templocation, templocation2);
					break;
					
				default:
					ProxyLogging.log(Level.MISHAP, "ProxyManager - ManagerConnectionProcessor -> "+peerID, "Unknown command: "+command[streamid]);
					// socket has read all of the data for this command, skip it now
				}
				packetdata[streamid]=null;
				streamid=-1;
			}
			if (dataread==-1)
			{
				ProxyLogging.log(Level.INFO, "ProxyManager - ManagerConnectionProcessor -> "+peerID, "Disconnecting, could not read data");
			}
		}
		catch (IOException e)
		{
			//e.printStackTrace();
			ProxyLogging.log(Level.INFO, "ProxyManager - ManagerConnectionProcessor -> "+peerID, "Disconnecting, error while reading data");
			if (peerID!=null)
				ManagerConnectionManager.getInstance().remove(peerID, this);
		}
		

		running=false;
		try {
			socket.close();
		} catch (IOException e) {
			// closing down
		}
	}
	
	public boolean isRunning()
	{
		return running;
	}
	public boolean hasPeerID()
	{
		return peerID!=null;
	}
	public int getPeerID()
	{
		return peerID;
	}
	
	private void sendInstanceID(OutputStream output) throws IOException
	{
		//ProxyLogging.log(Level.DEBUG, "ProxyManager - ManagerConnectionProcessor -> "+peerID, "Sending peer id "+Main.InstanceID);
		NetworkWriter net=new NetworkWriter(streams);
		net.writeInt(Main.InstanceID);
		net.flush(CMD_PEERID);
	}
	public void sendManagerLocation(ServiceLocation manager)
	{
		try {
			//ProxyLogging.log(Level.DEBUG, "ProxyManager - ManagerConnectionProcessor -> "+peerID, "Sending manager location "+manager);
			NetworkWriter net=new NetworkWriter(streams);
			net.writeServiceLocation(manager);
			net.flush(CMD_FOUNDMANAGER);
		}
		catch (IOException e)
		{
			ManagerConnectionManager.getInstance().remove(peerID, this);
		}
	}
	public void openCallbackPort(ServiceLocation local) throws IOException
	{
		try {
			//ProxyLogging.log(Level.DEBUG, "ProxyManager - ManagerConnectionProcessor -> "+peerID, "Sending request for callback port: "+local);
			NetworkWriter net=new NetworkWriter(streams);
			net.writeServiceLocation(local);
			net.flush(CMD_OPENCALLBACK);
		} catch (IOException e) {
			ManagerConnectionManager.getInstance().remove(peerID, this);
			throw new IOException("Failed to communicate with peer");
		}
	}
	
	/**
	 * After a callback port has been requested and established,
	 * inform the requester about it
	 * @param destination The destination port that the requester wanted to have called back
	 * @param reply The port that it is established at
	 * @throws IOException
	 */
	private void replyCallbackPort(ServiceLocation destination, ServiceLocation reply) throws IOException
	{
		try {
			//ProxyLogging.log(Level.DEBUG, "ProxyManager - ManagerConnectionProcessor -> "+peerID, "Sending reply for callback port, mapping "+reply+" to "+destination);
			NetworkWriter net=new NetworkWriter(streams);
			net.writeServiceLocation(destination);
			net.writeServiceLocation(reply);
			net.flush(CMD_REPLYCALLBACK);
		} catch (IOException e) {
			ManagerConnectionManager.getInstance().remove(peerID, this);
			throw new IOException("Failed to communicate with peer");
		}
	}
	
	/**
	 * If a callback port has been requested but failed,
	 * inform the requester about it
	 * @param destination The destination port that the requester wanted to have called back
	 * @throws IOException
	 */
	private void errorCallbackPort(ServiceLocation destination) throws IOException
	{
		try {
			//ProxyLogging.log(Level.DEBUG, "ProxyManager - ManagerConnectionProcessor -> "+peerID, "Sending error for callback port, requested "+destination);
			NetworkWriter net=new NetworkWriter(streams);
			net.writeServiceLocation(destination);
			net.flush(CMD_ERRORCALLBACK);
		} catch (IOException e) {
			ManagerConnectionManager.getInstance().remove(peerID, this);
			throw new IOException("Failed to communicate with peer");
		}
	}
	
	/**
	 * Send data from this a client port to the desired destination port
	 * @param from	Client port
	 * @param to	The actual port on the remote system
	 * @param data
	 * @throws IOException 
	 */
	public void sendData(ServiceLocation from, ServiceLocation to, byte[] data, int size) throws IOException
	{
		try {
			//ProxyLogging.log(Level.DEBUG, "ProxyManager - ManagerConnectionProcessor -> "+peerID, "Sending "+size+" data from "+from+" to "+to);
			NetworkWriter net=new NetworkWriter(streams);
			net.writeServiceLocation(from);
			net.writeServiceLocation(to);
			net.writeBytes(data, size);
			net.flush(CMD_SENDDATA);
		} catch (IOException e) {
			ManagerConnectionManager.getInstance().remove(peerID, this);
			throw new IOException("Failed to communicate with peer");
		}
	}
	
	/**
	 * Return a response to a SendData call
	 * @param from		Where the data is coming from, the server port
	 * @param to		Where the data is going, the client port
	 * @param data		The data
	 * @param size		How much of the data to send
	 * @throws IOException
	 */
	public void returnData(ServiceLocation from, ServiceLocation to, byte[] data, int size) throws IOException
	{
		try {
			//ProxyLogging.log(Level.DEBUG, "ProxyManager - ManagerConnectionProcessor -> "+peerID, "Returning "+size+" data from "+from+" to "+to);
			NetworkWriter net=new NetworkWriter(streams);
			net.writeServiceLocation(from);
			net.writeServiceLocation(to);
			net.writeBytes(data, size);
			net.flush(CMD_RETURNDATA);
		} catch (IOException e) {
			ManagerConnectionManager.getInstance().remove(peerID, this);
			throw new IOException("Failed to communicate with peer");
		}
	}

	/**
	 * Close the data connection from this client port to the given destination port
	 * @param from	Client port
	 * @param to	Server port
	 * @param data
	 * @throws IOException 
	 */
	public void closeDataConnection(ServiceLocation from, ServiceLocation to) throws IOException
	{
		try {
			//ProxyLogging.log(Level.DEBUG, "ProxyManager - ManagerConnectionProcessor -> "+peerID, "Closing data connection from "+from+" to "+to);
			NetworkWriter net=new NetworkWriter(streams);
			net.writeServiceLocation(from);
			net.writeServiceLocation(to);
			net.flush(CMD_CLOSEDATA);
		} catch (IOException e) {
			ManagerConnectionManager.getInstance().remove(peerID, this);
			throw new IOException("Failed to communicate with peer");
		}
	}
	
	@Override
	public String toString()
	{
		return "ManagerConnectionProcessor: "+peerID+"@"+socket.getRemoteSocketAddress();
	}
}
